package org.example.rabbitmq.consumer.mq;

import org.example.rabbitmq.consumer.rate.RateLimiter;
import org.example.rabbitmq.consumer.repository.QueueRepository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.SmartLifecycle;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.locks.ReentrantLock;

/**
 * RabbitMq队列管理帮助类
 */
public class RabbitMqManager implements SmartLifecycle {
    Logger logger = LoggerFactory.getLogger(RabbitMqManager.class);
    private static Set<String> queuesInThisAppInstance = new HashSet<String>();
    private List<SimpleMessageListenerContainer> containers = new ArrayList<>();
    ReentrantLock lock = new ReentrantLock();
    @Autowired
    ConnectionFactory connectionFactory;
    @Autowired
    QueueRepository queueRepository;
    @Autowired
    RateLimiter rateLimiter;

    ChannelAwareMessageListener messageListener;
    boolean isRunning = false;

    public RabbitMqManager(ChannelAwareMessageListener messageListener) {
        this.messageListener = messageListener;
    }

    public boolean createQueueContainerIfAbsent(String queueName) {
        // queue可以反复创建，但是没有必要每次发送都创建
        lock.lock();
        try {
            if (queuesInThisAppInstance.contains(queueName)) {
                logger.info("queues already have listening container, queue: " + queueName);
                return true;
            }
            // 动态创建Queue监听
            //  由于 MessageListener相同，可以用一个Container监听
            registerContainer(queueName, messageListener, AcknowledgeMode.MANUAL);
            // 注册队列限流器
            rateLimiter.addQueueRateLimiter(queueName);
            queuesInThisAppInstance.add(queueName);
            logger.info("dynamic create queue listening container success, queue: " + queueName);
            logger.info("queue listening container size: " + containers.size() + ", queue size: " + containers.size());
            return true;
        } catch (Exception ex) {
            logger.error("dynamic create queue listening container failed, queue: " + queueName, ex);
            return false;
        } finally {
            lock.unlock();
        }
    }

    @Deprecated
    private boolean createQueueContainerIfAbsent(String... queueNames) {
        // queue可以反复创建，但是没有必要每次发送都创建
        lock.lock();
        try {
            List<String> existQueueNames = new ArrayList<>();
            List<String> unListenedQueueNames = new ArrayList<>();
            for (String queueName : queueNames) {
                if (queuesInThisAppInstance.contains(queueName)) {
                    existQueueNames.add(queueName);
                } else {
                    unListenedQueueNames.add(queueName);
                }
            }
            if (existQueueNames.size() > 0) {
                logger.info("queues already have listening container, queues: " + String.join(",", queueNames));
            }
            // 动态创建Queue监听
            if (unListenedQueueNames.size() > 0) {
                //  由于 MessageListener相同，可以用一个Container监听
                registerContainer(unListenedQueueNames.toArray(new String[unListenedQueueNames.size()]), messageListener, AcknowledgeMode.MANUAL);
                for (String queueName : unListenedQueueNames) {
                    queuesInThisAppInstance.add(queueName);
                }
            }
            logger.info("dynamic create queue listening container success, queues: " + String.join(",", queueNames));
            Integer queueSize = containers.stream().map(c -> c.getQueueNames().length).reduce((x, y) -> x + y).orElse(0);
            logger.info("queue listening container size: " + containers.size() + ", queue size: " + queueSize);
            return true;
        } catch (Exception ex) {
            logger.error("dynamic create queue listening container failed, queues: " + String.join(",", queueNames), ex);
            return false;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 注册消息监听者
     *
     * @param queueName       消协队列名称
     * @param messageListener 消息处理程序{@link MessageListener}
     * @param mode            消息确认机制，消息处理程序为{@link ChannelAwareMessageListener}类型时，使用{@link AcknowledgeMode#MANUAL}
     */
    private void registerContainer(String queueName, MessageListener messageListener, AcknowledgeMode mode) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueueNames(queueName);
        container.setMessageListener(messageListener);
        container.setAcknowledgeMode(mode);
        // 如果要求一个队列同一时间只有一个consumer处理消息
        // prefetchCount, batchSize 每次取多少个, 是多少都没有影响，与需求无关
        // exclusive需要为true, 连带concurrentConsumers和maxConcurrentConsumers为1
        //
        // 连带引出需要每个队列，都对应一个Container，所以queueNames只能有一个
//        container.setPrefetchCount(1); // how many messages to send to each consumer in a single request
        container.setExclusive(true); // concurrentConsumers == 1 + maxConcurrentConsumers == 1
//        container.setConcurrentConsumers(1); // <= maxConcurrentConsumers
//        container.setMaxConcurrentConsumers(1); // >= concurrentConsumers
//        container.setBatchSize(1); // <= PrefetchCount ,AcknowledgeMode#AUTO, setConsumerBatchEnabled+setReceiveTimeout,setDeBatchingEnabled
        containers.add(container);
        if (!container.isRunning()) {
            container.start();
        }
    }

    /**
     * 注册消息监听者
     *
     * @param queueNames      消协队列名称，可以监听多个消息队列
     * @param messageListener 消息处理程序{@link MessageListener}
     * @param mode            消息确认机制，消息处理程序为{@link ChannelAwareMessageListener}类型时，使用AcknowledgeMode's {@code MANUAL 手动确认}
     */
    @Deprecated
    private void registerContainer(String[] queueNames, MessageListener messageListener, AcknowledgeMode mode) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueueNames(queueNames);
        container.setMessageListener(messageListener);
        container.setAcknowledgeMode(mode);
        containers.add(container);
        if (!container.isRunning()) {
            container.start();
        }
    }

    private void loadQueuesOnStart() {
        Set<String> queues = null;
        try {
            queues = queueRepository.getQueueNames();
        } catch (Exception ex) {
            logger.error("load queue names from redis failed", ex);
        }
        if (queues != null && queues.size() > 0) {
            // createQueueContainerIfAbsent(queues.toArray(new String[queues.size()]));
            for (String queue : queues) {
                createQueueContainerIfAbsent(queue);
            }
        }
    }

    private void startMessageListenerContainers() {
        for (SimpleMessageListenerContainer container : containers) {
            if (!container.isRunning()) {
                try {
                    container.start();
                    logger.error("start queue listening container success, queues: " + String.join(",", container.getQueueNames()));
                } catch (Exception ex) {
                    logger.error("start queue listening container failed, queues: " + String.join(",", container.getQueueNames()), ex);
                }
            }
        }
    }

    private void stopMessageListenerContainers() {
        for (SimpleMessageListenerContainer container : containers) {
            if (container.isRunning()) {
                try {
                    container.stop();
                    logger.error("stop queue listening container success, queues: " + String.join(",", container.getQueueNames()));
                } catch (Exception ex) {
                    logger.error("stop queue listening container failed, queues: " + String.join(",", container.getQueueNames()), ex);
                }
            }
        }
    }

    @Override
    public void start() {
        isRunning = true;
        lock.lock();
        try {
            // 从Redis中加载已经注册的Queue
            loadQueuesOnStart();
            // 启动Container
            startMessageListenerContainers();
        } finally {
            lock.unlock();
        }
    }

    @Override
    public void stop() {
        isRunning = false;
        lock.lock();
        try {
            stopMessageListenerContainers();
        } finally {
            lock.unlock();
        }
    }

    @Override
    public boolean isRunning() {
        return isRunning;
    }
}